package org.mozhu.simple.rpc.server;

import lombok.extern.slf4j.Slf4j;
import org.mozhu.simple.rpc.server.network.PlaintextChannelBuilder;
import org.mozhu.simple.rpc.server.network.RequestChannel;
import org.mozhu.simple.rpc.server.network.RpcChannel;
import org.mozhu.simple.rpc.server.network.Selector;

import java.nio.channels.SocketChannel;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
public class Processor extends AbstractServerThread {

    private ConcurrentLinkedQueue<SocketChannel> newConnections = new ConcurrentLinkedQueue<SocketChannel>();
    private Selector selector = new Selector(new PlaintextChannelBuilder(), Integer.MAX_VALUE);

    private int id;
    private RequestChannel requestChannel;

    public Processor(int id, RequestChannel requestChannel) {
        this.id = id;
        this.requestChannel = requestChannel;
    }

    @Override
    public void wakeup() {

    }

    @Override
    public void run() {
        startupComplete();
        try {
            // setup any new connections that have been queued up
            configureNewConnections();

            // register any new responses for writing
            processNewResponses();
        } finally {
            shutdownComplete();
        }
    }

    private void processNewResponses() {
        RequestChannel.Response curr = requestChannel.receiveResponse(id);
        while (curr != null) {
            try {
                RequestChannel.ResponseAction responseAction = curr.getResponseAction();
                switch (responseAction) {
                    case NoOpAction:
                    case SendAction:
                        sendResponse(curr);
                    case CloseConnectionAction:
                        close(selector, curr.getRequest().getConnectionId());
                    default:
                }
            } finally {
                curr = requestChannel.receiveResponse(id);
            }
        }
    }

    protected void sendResponse(RequestChannel.Response response) {
        RpcChannel channel = selector.channel(response.getSend().destination());
        if (channel != null) {
            selector.send(response.getSend());
        }
    }

    private void configureNewConnections() {
        while (!newConnections.isEmpty()) {
            SocketChannel channel = newConnections.poll();
            try {
                String connectionId = UUID.randomUUID().toString();
                selector.register(connectionId, channel);
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    public void accept(SocketChannel socketChannel) {
        //TODO accept
    }

    public int getId() {
        return id;
    }
}
